package com.example.study.commen.cdc;


import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Properties;

/**
 * @author lei
 * @create 2022-08-25 13:42
 * @desc mysql变更监听
 **/
//@Component
public class MysqlEventListener implements ApplicationRunner {

    private final DataChangeSink dataChangeSink;

    public MysqlEventListener(DataChangeSink dataChangeSink) {
        this.dataChangeSink = dataChangeSink;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DebeziumSourceFunction<DataChangeInfo> dataChangeInfoMySqlSource = buildDataChangeSource();

        DataStream<DataChangeInfo> streamSource = env
                .addSource(dataChangeInfoMySqlSource, "mysql-source")
                .setParallelism(1);
        streamSource.addSink(dataChangeSink);
        dss(env);
        env.execute("mysql-stream-cdc");

    }

    private void dss(StreamExecutionEnvironment env) {
        env.enableCheckpointing(1000L * 60 * 60 * 12);//12小时
//        env.enableCheckpointing(1000L * 20);

        // 存储位置，FileSystemCheckpointStorage(文件存储)
//        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///c:/cdc/checkpoint/"));
        // savepoint存储位置
//        env.setDefaultSavepointDirectory("file:///c:/cdc/checkpoint/");
        // 超时时间，checkpoint没在时间内完成则丢弃
        env.getCheckpointConfig().setCheckpointTimeout(1000L * 10);
        // 同时并发数量
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        // 最小间隔时间（前一次结束时间，与下一次开始时间间隔）
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        // 外部checkpoint(例如文件存储)清除策略
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        env.setStateBackend(new FsStateBackend(""));
        env.setStateBackend(new MemoryStateBackend());
//        env.setStateBackend( new FsStateBackend("file:///data/flink/checkpoints"));
        ;
        //2.2 指定CK的一致性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //2.3 设置任务关闭的时候保留最后一次CK数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 指定从CK自动重启策略
        // 设置重启次数为3，重启间隔时间为1000*10ms
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000L * 10));

        //2.5 设置状态后端

//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));

        //2.6 设置访问HDFS的用户名

//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //3.创建Flink-MySQL-CDC的Source


        Properties debeziumProperties = new Properties();
    }


    /**
     * 构造变更数据源
     *
     * @param
     * @return DebeziumSourceFunction<DataChangeInfo>
     * @author lei
     * @date 2022-08-25 15:29:38
     */
    private DebeziumSourceFunction<DataChangeInfo> buildDataChangeSource() {
        return MySqlSource.<DataChangeInfo>builder()
                .hostname("8.130.21.209")
                .port(3306)
                .databaseList("study")
                .tableList("study.token_user", "study.user")
                .username("canal")
                .password("canal")

                /**initial初始化快照,即全量导入后增量导入(检测更新数据写入)
                 * latest:只进行增量导入(不读取历史变化)
                 * timestamp:指定时间戳进行数据导入(大于等于指定时间错读取数据)
                 */
                .startupOptions(StartupOptions.latest())
                .deserializer(new MysqlDeserialization())
                .serverTimeZone("GMT+8")
                .build();
    }
}

